package org.itew.mymapreduce.core.impl;

import java.io.IOException;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import org.itew.mymapreduce.core.DataCollection;
import org.itew.mymapreduce.core.Partition;
import org.itew.mymapreduce.util.PropertiesConfigurationReader;

public class HashPartition<K, V> implements Partition<K, V> {

	private int keySetSize;

	private Set<K> allKeySet;

	private Iterator<DataCollection<K, V>> dataCollectionIterator;

	private Iterator<K> kIterator;

	private int currentKeySetSize;

	@Override
	public void setData(Collection<DataCollection<K, V>> dataCollection,
			int reducerNumber) throws IOException {
		PropertiesConfigurationReader pcr = new PropertiesConfigurationReader(
				HashPartition.class, "hashpartition.properties");
		keySetSize = pcr.getInt("keySetSize");
		this.allKeySet = new HashSet<K>();
		dataCollectionIterator = dataCollection.iterator();
	}

	@Override
	public Set<K> nextKeySet() {

		Set<K> returnSetKey = new HashSet<K>();

		for (currentKeySetSize = 0; currentKeySetSize < keySetSize;) {
			if (kIterator == null || !kIterator.hasNext()) {
				if (!dataCollectionIterator.hasNext()) {
					if (returnSetKey.isEmpty())
						return null;
					else
						return returnSetKey;
				} else
					kIterator = dataCollectionIterator.next().keySet()
							.iterator();
			} else {
				K key = kIterator.next();
				if (!allKeySet.contains(key)) {
					returnSetKey.add(key);
					allKeySet.add(key);
					currentKeySetSize++;
				}
			}
		}
		
		return returnSetKey;
	}

}
